package cn.doitedu.demo2;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/1/21
 * @Desc: 学大数据，上多易教育
 * 针对  活跃等级为 C 的用户
 * 在规则上线后，如果这些用户 发生了 X行为 >=3 次
 * 等他再发生  Y行为的时候，触发营销消息
 * <p>
 * <p>
 * 针对  价值等级为 A 的用户
 * 在规则上线后，如果这些用户 发生了 W行为 >=2 次
 * 等他再发生 K行为的时候，触发营销消息
 **/
public class Demo2 {

    final static Logger logger = LoggerFactory.getLogger(Demo2.class.getName());

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        /* *
         * 用 cdc连接器去读取规则元数据表的变更信息
         *
         */
        MySqlSource<String> cdcSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .username("root")
                .password("root")
                .databaseList("doit44")
                .tableList("doit44.rule_meta")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        DataStreamSource<String> ruleMetaCdcStream = env.fromSource(cdcSource, WatermarkStrategy.noWatermarks(), "rule-meta");
        // 解析cdc-json，得到RuleMeta对象
        SingleOutputStreamOperator<RuleMeta> ruleMetaBeanStream = ruleMetaCdcStream.map(new MapFunction<String, RuleMeta>() {
            @Override
            public RuleMeta map(String cdcJson) throws Exception {

                JSONObject jsonObject = JSON.parseObject(cdcJson);
                String op = jsonObject.getString("op");

                JSONObject dataObject;
                if (op.equals("d")) {
                    dataObject = jsonObject.getJSONObject("before");
                } else {

                    dataObject = jsonObject.getJSONObject("after");
                }

                String ruleId = dataObject.getString("rule_id");
                String ruleParam = dataObject.getString("rule_param");
                int status = dataObject.getIntValue("status");

                return new RuleMeta(ruleId, ruleParam, status, op);
            }
        });

        MapStateDescriptor<String, String> bcStateDesc = new MapStateDescriptor<>("bc_state", String.class, String.class);
        // 将规则元数据对象流，广播，得到广播流
        BroadcastStream<RuleMeta> ruleMetaBroadcastStream = ruleMetaBeanStream.broadcast(bcStateDesc);



        /* *
         * 用 kafka连接器去读取行为日志流
         */
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setStartingOffsets(OffsetsInitializer.latest())
                .setGroupId("ggg111")
                .setClientIdPrefix("ccc111")
                .setTopics("dwd-events")
                .build();


        DataStreamSource<String> streamSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");

        SingleOutputStreamOperator<UserEvent> eventStream = streamSource.map(json -> JSON.parseObject(json, UserEvent.class));


        KeyedStream<UserEvent, Long> userEventKeyedStream = eventStream.keyBy(UserEvent::getUser_id);


        /* *
         *  userEvent keyed流  连接   规则元数据Bean 广播流
         */
        BroadcastConnectedStream<UserEvent, RuleMeta> connectedSteam = userEventKeyedStream.connect(ruleMetaBroadcastStream);


        /* *
         *  对 连接流，应用 process算子，进行核心逻辑处理
         */

        SingleOutputStreamOperator<Message> messages = connectedSteam.process(new KeyedBroadcastProcessFunction<Long, UserEvent, RuleMeta, Message>() {


            HashMap<String, RuleCalculator> calculatorPool;

            @Override
            public void open(Configuration parameters) throws Exception {
                calculatorPool = new HashMap<>();
            }

            /**
             *  处理 行为事件流中的数据 UserEvent
             */
            @Override
            public void processElement(UserEvent userEvent, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMeta, Message>.ReadOnlyContext ctx, Collector<Message> out) throws Exception {

                for (Map.Entry<String, RuleCalculator> entry : calculatorPool.entrySet()) {
                    RuleCalculator ruleCalculator = entry.getValue();
                    ruleCalculator.calc(userEvent, out);
                }

            }

            /**
             * 处理 广播流中的数据  RuleMeta
             */
            @Override
            public void processBroadcastElement(RuleMeta ruleMeta, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMeta, Message>.Context ctx, Collector<Message> out) throws Exception {

                String op = ruleMeta.getOp();
                String ruleId = ruleMeta.getRule_id();
                int status = ruleMeta.getStatus();

                // 如果op是新增、或更新，且最终status为1，表示要新上线一个规则
                if ((op.equals("c") && status == 1) || (op.equals("u") && status == 1)  || (op.equals("r") && status ==1 )) {
                    // 构造一个新的运算机对象
                    RuleCalculator ruleCalculator = new RuleModel_1_Calculator();
                    /**
                     * TODO :
                     *  1. 从 ruleMeta 中 get 模型运算机类的源代码
                     *  2. 编译源代码
                     *  3.加载 class到 jvm
                     *  4. 反射 class得到运算机对象
                     */


                    // 初始化运算机对象
                    ruleCalculator.init(getRuntimeContext(), ruleMeta.getRule_param());
                    // 将初始化好的运算机对象，放入运算机池
                    calculatorPool.put(ruleId, ruleCalculator);

                    logger.warn("新上线了一个规则,规则id:{}", ruleId);

                }
                // 如果op是删除，或者是更新且status更新成了 ！= 1，则表示要下线一个规则
                else if (op.equals("d") || (op.equals("u") && status != 1)) {
                    calculatorPool.remove(ruleId);
                    logger.warn("下线了一个规则,规则id:{}", ruleId);

                } else {
                    logger.error("本次规则操作不合法:{}", ruleMeta);
                }


            }
        });


        messages.print();

        env.execute();

    }
}
